﻿using Confluent.Kafka;
using System;
using System.Threading.Tasks;

namespace Transactions
{
	enum ProducerState
	{
		MakingMessagesToAbort,
		MakingMessagesToCommit,
		InitState
	}
	class Program
	{
		static TimeSpan DefaultTimeout = TimeSpan.FromSeconds(3);
		static void Main(string[] args)
		{
			string brokerList = "39.96.82.51:9093,47.95.2.2:9092,39.96.34.52:9092";
			string topicName = "test";
			string transactionalId = "testtransactionalId";
			var config = new ProducerConfig
			{
				BootstrapServers = brokerList,
				//幂等性
				EnableIdempotence = true,
				Acks = Acks.All,
				//LingerMs = 10000,
				//BatchNumMessages=1,
				//MessageSendMaxRetries = 3,
				TransactionalId = transactionalId,
				 

			
				//Partitioner = Partitioner.Murmur2Random

			};

			using (var producer = new ProducerBuilder<string, string>(config)
				.Build())
			{
				try
				{
					producer.InitTransactions(DefaultTimeout);
					var currentState = ProducerState.InitState;
					producer.BeginTransaction();
					for (int i = 100; i < 110; i++)
					{
						var content = i.ToString();
						producer.Produce(
						   topicName, new Message<string, string> { Key = content, Value = content });
					}
					Console.WriteLine("输入");
					Console.ReadLine();
					//var content1 = "107";
					//producer.Produce(
					//	   topicName, new Message<string, string> { Key = content1, Value = content1 });

					producer.CommitTransaction(DefaultTimeout);
				}
				catch (Exception ex)
				{
					producer.AbortTransaction(DefaultTimeout);
					Console.WriteLine(ex.Message);
				}



				Console.WriteLine($"done: {transactionalId}");
			}
		}
	}
}
